package com.codejiwei.flink.cdc;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.ConnectSchema;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.json.JsonConverter;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.storage.ConverterType;

import java.util.Base64;
import java.util.HashMap;
import java.util.List;

/**
 * author: codejiwei
 * date: 2023/8/27
 * desc:
 **/
public class CustomStringDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    private static final long serialVersionUID = 1L;
    private transient JsonConverter jsonConverter;
    private final Boolean includeSchema;
    private transient Base64.Decoder decoder;
    private transient ObjectMapper objectMapper;

    public CustomStringDebeziumDeserializationSchema() {
        this(false);
    }

    public CustomStringDebeziumDeserializationSchema(Boolean includeSchema) {
        this.includeSchema = includeSchema;
    }

    @Override
    public void deserialize(SourceRecord record, Collector<String> out) throws Exception {

        if (this.jsonConverter == null) {
            this.initializeJsonConverter();
        }
        if (this.decoder == null) {
            this.decoder = Base64.getDecoder();
        }
        if (this.objectMapper == null) {
            this.objectMapper = new ObjectMapper();
        }
        String binlog = new String(this.jsonConverter.fromConnectData(record.topic(), record.valueSchema(), record.value()));
        System.out.println("===============record.valueSchema: " + record.valueSchema());
        System.out.println("===============record.key: " + record.key());

        ConnectSchema schema = (ConnectSchema) record.valueSchema();
        List<Field> fields = schema.fields();
        System.out.println("------------------fields: " + fields.toString());


        out.collect(binlog);

    }

    private void initializeJsonConverter() {
        this.jsonConverter = new JsonConverter();
        HashMap<String, Object> configs = new HashMap<>(2);
        configs.put("converter.type", ConverterType.VALUE.getName());
        configs.put("schemas.enable", this.includeSchema);
        this.jsonConverter.configure(configs);
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.of(String.class);
    }
}
